package window

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
import org.apache.flink.util.Collector

import java.text.SimpleDateFormat
import scala.util.Random

object WindowDemo {

  /*def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val data: DataStream[String] = env.socketTextStream("linux121", 7777)
    val sdf = new SimpleDateFormat("HH:mm:ss.SSS")

    implicit val typeInfo = TypeInformation.of(classOf[(String, String, String)])
    val maped: DataStream[(String, String, String)] = data.map((value: String) => {
      val l = System.currentTimeMillis
      val dataTime = sdf.format(l)
      val randomNum = Random.nextInt(5)
      (value, dataTime, String.valueOf(randomNum))
    })

    maped.print()
    implicit val typeInfo2 = TypeInformation.of(classOf[String])
    val keyed: KeyedStream[(String, String, String), String] = maped.keyBy((value: (String, String, String)) => value._1)
    val countWindow: WindowedStream[(String, String, String), String, GlobalWindow] = keyed.countWindow(3, 1)

   // implicit val typeInfo3 = TypeInformation.of(classOf[String])

    val applyed = countWindow.apply(new WindowFunction[(String, String, String), String, String, GlobalWindow] {

      override def apply(key: String, window: GlobalWindow, input: Iterable[(String, String, String)], out: Collector[String]): Unit = {
        val sb = new StringBuilder()
        input.foreach(it => {
          val next = it
          sb.append(next._1 + ".." + next._2 + ".." + next._3)
          out.collect(sb.toString())
        })
      }
    })

    applyed.print()
    env.execute()

  }*/

}
